package com.zhuchang.self_rpc.test.server;

import com.zhuchang.self_rpc.test.client.RpcClientHandler;
import com.zhuchang.self_rpc.test.codec.RpcDencoder;
import com.zhuchang.self_rpc.test.codec.RpcEncoder;
import com.zhuchang.self_rpc.test.codec.RpcRequest;
import com.zhuchang.self_rpc.test.codec.RpcResponse;
import com.zhuchang.self_rpc.test.config.ProviderConfig;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * @Author RenPu
 * @Date 2021/8/18 14:34
 * @Version 1.0
 * @Description: 基于netty自定义RPC框架: 服务端的编写
 **/


@Slf4j
public class RpcServer {

    //ip+端口的
    private String serverAddress;

    //用于接受连接组
    private NioEventLoopGroup bossGroup = new NioEventLoopGroup();

    //实际处理业务连接组
    private NioEventLoopGroup workGroup = new NioEventLoopGroup();

    // key:interfaceName（接口名称）,value: interfaceImpl(实现)
    private Map<String, Object> handlerMap = new HashMap<String, Object>();


    public RpcServer(String serverAddress) throws InterruptedException {
        this.serverAddress = serverAddress;
        this.start();
    }


    /**
     * 启动方法
     */
    public void start() throws InterruptedException {

        //创建服务端启动对象
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.group(bossGroup, workGroup)                 //绑定接收连接组+具体处理业务组
                .channel(NioServerSocketChannel.class)             //
                .option(ChannelOption.SO_BACKLOG, 1024)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel cp) throws Exception {
                        //实际编解码
                        ChannelPipeline pipeline = cp.pipeline();

                        //netty提供的编码器处理，65336：整个数据包的大小设置，0：起始，4：包头
                        pipeline.addLast(new LengthFieldBasedFrameDecoder(65336, 0, 4, 0, 0));

                        //对请求对象进行编码
                        pipeline.addLast(new RpcEncoder(RpcResponse.class));

                        //对响应对象进行解码
                        pipeline.addLast(new RpcDencoder(RpcRequest.class));

                        pipeline.addLast(new RpcServerHandler(handlerMap));
                    }
                });

        //解析出ip+port
        String[] split = serverAddress.split(":");
        String host = split[0];
        int port = Integer.parseInt(split[1]);

        //连接触发启动 sync 同步
        ChannelFuture channelFuture = serverBootstrap.bind(host, port).sync();

        //添加异步监听事件，观察对应的状态
        channelFuture.addListener(new ChannelFutureListener() {

            public void operationComplete(ChannelFuture future) throws Exception {
                if (future.isSuccess()) {
                    log.info("server success  bing to :" + serverAddress);
                } else {
                    log.info("server fail  bing to :" + serverAddress);
                    throw new Exception("server start fail cause:" + future.cause());
                }

            }
        });


        //同步定时阻塞
        try {
            channelFuture.await(5000, TimeUnit.MICROSECONDS);
            if (channelFuture.isSuccess()) {
                log.info("start rapid rpc success ！！！");
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
            log.info("start rapid rpc occur Interrupted, ex:" + e);
        }


    }


    /**
     * 程序注册器方法
     */
    public void registerProcessor(ProviderConfig providerConfig) {

        //key:接口的全限定名称，value: 接口的实现类的实例
        handlerMap.put(providerConfig.getInterfaceClass(), providerConfig.getRef());

    }


    /**
     * 资源释放方法
     */
    public void close() {
        bossGroup.shutdownGracefully();
        workGroup.shutdownGracefully();

    }


}
